/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.
 */

package com.huaweicloud.reqmt;

import com.huaweicloud.reqmt.domain.dwd_haydn_case_uuid_vaild_dura_stage_d;
import com.huaweicloud.reqmt.function.DwdMapFunction;
import com.huaweicloud.reqmt.source.DwsSourceDwd001;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 代码 DWD001
 * 表名 dwd_haydn_case_uuid_vaild_dura_stage_d
 *
 * @since 2022-11-03
 */
public class MainDwd001 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String username = args[0];
        String password = args[1];
        String url = args[2];
        // 添加数据源
        DataStreamSource<dwd_haydn_case_uuid_vaild_dura_stage_d> source =
            env.addSource(new DwsSourceDwd001(username, password, url));
        // 排除节假日
        SingleOutputStreamOperator<dwd_haydn_case_uuid_vaild_dura_stage_d> data = source.map(new DwdMapFunction());
        data.print();
        // addSink 将数据导回DWS
        data.addSink(JdbcSink.sink(
            "insert into dwd.dwd_haydn_case_uuid_valid_dura_stage_noh_d VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (ps, table) -> {
                ps.setString(1, table.getDs());
                ps.setString(2, table.getWorkspaceId());
                ps.setString(3, table.getWorkspaceName());
                ps.setString(4, table.getSolutionUuid());
                ps.setString(5, table.getSoluName());
                ps.setString(6, table.getSoluType());
                ps.setString(7, table.getReqmtUuid());
                ps.setString(8, table.getReqmtName());
                ps.setString(9, table.getStage());
                ps.setString(10, table.getcStage());
                ps.setString(11, table.getDefinition());
                ps.setString(12, table.getStartTime());
                ps.setString(13, table.getEndTime());
                ps.setString(14, table.getDura());
            },
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl(url)
                .withDriverName("org.postgresql.Driver")
                .withUsername(username)
                .withPassword(password)
                .build()));
        // 执行
        env.execute("dwd_haydn_case_uuid_valid_dura_stage_noh_d");
    }
}
